package com.xiaoxu.spark.DateFrameWithDataset

import org.apache.spark.sql.SparkSession
import org.slf4j.{Logger, LoggerFactory}


object DataFrameCase {


  private val logger: Logger = LoggerFactory.getLogger(DataFrameCase.getClass())

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("DataFrameRDDApp")
      .master("local[6]")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.scheduler.mode", "FAIR")
      .getOrCreate()

    /*logger.trace("trace")
    logger.debug("debug")
    logger.info("info")
    logger.warn("warn")
    logger.error("error")*/
    // RDD ==> DataFrame
    val rdd = spark.sparkContext.textFile("data/student.data")


    //注意：需要导入隐式转换
    import spark.implicits._
    // 注意添加转义字符
    val studentDF = rdd
      .map(_.split(","))
      .map(line => Student(line(0).toInt, line(1), line(2), line(3))).toDF()

    //show默认只显示前20条
    //studentDF.show
    //studentDF.show(30)
    //truncate = false 不会截取超过长度的字符串，字符串会全部显示出来
   // studentDF.show(30, false)

    //返回前十条数据
    //studentDF.take(10)
    //返回第一行数据
    //studentDF.first()
    //返回前三行数据
    //studentDF.head(3)


    //studentDF.select("email").show(30, false)


    //studentDF.filter("name=''").show
    //studentDF.filter("name='' OR name='NULL'").show


    //name以M开头的人
    //studentDF.filter("SUBSTR(name,0,1)='M'").show

    //排序
    //studentDF.sort(studentDF("name")).show
    //studentDF.sort(studentDF("name").desc).show

    //studentDF.sort("name", "id").show
    //studentDF.sort(studentDF("name").asc, studentDF("id").desc).show

    //重命名
    //studentDF.select(studentDF("name").as("student_name")).show


    //连接
    val studentDF2 = rdd
      .map(_.split(","))
      .map(line => Student(line(0).toInt, line(1), line(2), line(3))).toDF()


    studentDF.join(studentDF2, studentDF.col("id") === studentDF2.col("id")).show

    spark.stop()
  }

  case class Student(id: Int, name: String, phone: String, email: String)


}
